/**
 * collectd - src/rrdcached.c
 * Copyright (C) 2008  Florian octo Forster
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; only version 2 of the License is applicable.
 *
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License along
 * with this program; if not, write to the Free Software Foundation, Inc.,
 * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
 *
 * Authors:
 *   Florian octo Forster <octo at verplant.org>
 **/
/**
   This is a plugin that uses mysql as storage backend!
**/

#include "collectd.h"
#include "plugin.h"
#include "common.h"
#include "configfile.h"

#include <string.h>
#include <stdlib.h>
#include <time.h>
#ifdef HAVE_MYSQL_H
#include <mysql.h>
#elif defined(HAVE_MYSQL_MYSQL_H)
#include <mysql/mysql.h>
#endif

struct mysql_database_s /* {{{ */
{
  /* instance == NULL  =>  legacy mode */
  char *host;
  char *user;
  char *pass;
  char *database;
  char *socket;
  int   port;

  MYSQL *con;
  int    state;
};
typedef struct mysql_database_s mysql_database_t; /* }}} */

static const char *mysql_config_keys[] =
{
  "Host", "User", "Password",
  "Database", "Socket", "Port"
};
static int mysql_config_keys_num = STATIC_ARRAY_SIZE (mysql_config_keys);

static mysql_database_t mydb;
static char schema_host[2048];
static const int MAX_QUERY_LEN = 4096;
static char query_str[4096];
static const int JOB_ID_LEN = 17;
static const int TASK_ID_LEN = 11;

static void mysql_database_free (mysql_database_t *db) /* {{{ */
{
  if (db == NULL)
    return;

  if (db->con != NULL)
    mysql_close (db->con);

  sfree (db->host);
  sfree (db->user);
  sfree (db->pass);
  sfree (db->socket);
  sfree (db->database);
} /* }}} void mysql_database_free */

/* Configuration handling functions {{{
 *
 * <Plugin mysql>
 *   <Database "plugin_instance1">
 *     Host "localhost"
 *     Port 22000
 *     ...
 *   </Database>
 * </Plugin>
 }}} */

static int mysql_config_set_string (char **ret_string, /* {{{ */
  const char *val)
{
  char *string;
  string = strdup (val);
  if (string == NULL)
  {
    ERROR ("mysql plugin: strdup failed.");
    return (-1);
  }
  if (*ret_string != NULL)
    free (*ret_string);
  *ret_string = string;
  return (0);
} /* }}} int mysql_config_set_string */

static int mysql_config_set_int (int *ret_int, /* {{{ */
  const char *val)
{
  if (sscanf(val, "%d", ret_int) != 0) {
    return -1;
  } else {
    return 0;
  }
} /* }}} int mysql_config_set_int */

static int mysql_collectd_config (const char *key, const char *value) /* {{{ */
{
  int status;
  /* Fill the `mysql_database_t' structure.. */
  if (strcasecmp ("Host", key) == 0)
    status = mysql_config_set_string (&(mydb.host), value);
  else if (strcasecmp ("User", key) == 0)
    status = mysql_config_set_string (&(mydb.user), value);
  else if (strcasecmp ("Password", key) == 0)
    status = mysql_config_set_string (&mydb.pass, value);
  else if (strcasecmp ("Port", key) == 0)
    status = mysql_config_set_int (&mydb.port, value);
  else if (strcasecmp ("Socket", key) == 0)
    status = mysql_config_set_string (&mydb.socket, value);
  else if (strcasecmp ("Database", key) == 0)
    status = mysql_config_set_string (&mydb.database, value);
  else
  {
    WARNING ("mysql plugin: Option `%s' not allowed here.", key);
    status = -1;
  }
  return status;
} /* }}} int mysql_config*/

static int value_list_to_string (char *buffer, int buffer_len,/*{{{*/
    const data_set_t *ds, const value_list_t *vl)
{
  int offset;
  int status;
  int i;
  assert (0 == strcmp (ds->type, vl->type));

  memset (buffer, '\0', buffer_len);
  offset = 0;
  for (i = 0; i < ds->ds_num; i++)
  {
    if ((ds->ds[i].type != DS_TYPE_COUNTER)
        && (ds->ds[i].type != DS_TYPE_GAUGE)
        && (ds->ds[i].type != DS_TYPE_DERIVE)
        && (ds->ds[i].type != DS_TYPE_ABSOLUTE))
      return (-1);

    if (ds->ds[i].type == DS_TYPE_COUNTER)
    {
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%llu", vl->values[i].counter);
    }
    else if (ds->ds[i].type == DS_TYPE_GAUGE)
    {
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%f", vl->values[i].gauge);
    }
/*
    else if (ds->ds[i].type == DS_TYPE_DERIVE) {
      status = ssnprintf (buffer + offset, buffer_len - offset,
    "%"PRIi64, vl->values[i].derive);
    }
    else if (ds->ds[i].type == DS_TYPE_ABSOLUTE) {
      status = ssnprintf (buffer + offset, buffer_len - offset,
    "%"PRIu64, vl->values[i].absolute);
    }
*/
    if ((status < 1) || (status >= (buffer_len - offset)))
      return (-1);

    offset += status;
  } /* for ds->ds_num */

  return (0);
} /* int value_list_to_string *//*}}}*/

void strcpyn(char *dst, const char *src, int len) {
  int i;
  for (i = 0; i < len; ++i) {
    dst[i] = src[i];
  }
}

static int value_list_to_tablename (char *buffer, int buffer_len,/*{{{*/
    const data_set_t *ds, const value_list_t *vl)
{
  int offset = 0;
  int status;

  assert (0 == strcmp (ds->type, vl->type));
  if (strcmp(vl->plugin, "mrjob_task") == 0 &&
      strlen(vl->plugin_instance) > 0) {
    char jobid[JOB_ID_LEN + 1];
    jobid[JOB_ID_LEN] = '\0';
    strcpyn(jobid, vl->plugin_instance, JOB_ID_LEN);
    status = ssnprintf (buffer + offset, buffer_len - offset,
                    "mrjob_%s", jobid);
    if ((status < 1) || (status >= buffer_len - offset))
            return (-1);
    offset += status;
  } else {
    if (strlen (vl->plugin_instance) > 0)
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s_", vl->plugin_instance);
    else
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s_", vl->plugin);
    if ((status < 1) || (status >= buffer_len - offset))
      return (-1);
    offset += status;

    if (strlen (vl->type_instance) > 0)
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s", vl->type_instance);
    else
      status = ssnprintf (buffer + offset, buffer_len - offset,
          "%s", vl->type);
    if ((status < 1) || (status >= buffer_len - offset))
      return (-1);
    offset += status;
  }
  buffer[buffer_len - 1] = 0;
  return (0);
} /* int value_list_to_filename *//*}}}*/

static MYSQL *getconnection (mysql_database_t *db)
{/*{{{*/
  if (db->state != 0)
  {
    int err;
    if ((err = mysql_ping (db->con)) != 0)
    {
      WARNING ("mysql_ping failed for %s",
               mysql_error (db->con));
      db->state = 0;
    }
    else
    {
      db->state = 1;
      return (db->con);
    }
  }

  if ((db->con = mysql_init (db->con)) == NULL)
  {
    ERROR ("mysql_init failed: %s", mysql_error (db->con));
    db->state = 0;
    return (NULL);
  }

  if (mysql_real_connect (db->con, db->host, db->user, db->pass,
      db->database, db->port, db->socket, 0) == NULL)
  {
    ERROR ("mysql plugin: Failed to connect to database %s "
           "at server %s: %s",
           (db->database != NULL) ? db->database : "<none>",
           (db->host != NULL) ? db->host : "localhost",
           mysql_error (db->con));
    db->state = 0;
    return (NULL);
  }
  else
  {
    INFO ("mysql plugin: Successfully connected to database %s "
      "at server %s (server version: %s, protocol version: %d)",
      (db->database != NULL) ? db->database : "<none>",
      mysql_get_host_info (db->con),
      mysql_get_server_info (db->con),
      mysql_get_proto_info (db->con));
    db->state = 1;
    return (db->con);
  }
} /* static MYSQL *getconnection (mysql_database_t *db) *//*}}}*/

static MYSQL_RES *exec_query (MYSQL *con, const char *query)
{/*{{{*/
  MYSQL_RES *res;

  int query_len = strlen (query);

  if (mysql_real_query (con, query, query_len))
  {
    ERROR ("mysql plugin: Failed to execute query: %s",
      mysql_error (con));
    INFO ("mysql plugin: SQL query was: %s", query);
    return (NULL);
  }

  res = mysql_store_result (con);

  return (res);
} /* exec_query *//*}}}*/

static int wrap_ssnprintf_str(char *buffer, size_t buffer_len, int *offset,
                          const char *arg, const char* val)
{/*{{{*/
  int status;
  status = ssnprintf (buffer + *offset, buffer_len - *offset, arg, val);
  if ((status < 1) || (status >= buffer_len - *offset))
     return (-1);
  *offset += status;
  return 0;
}/*}}}*/

static int wrap_ssnprintf_int(char *buffer, size_t buffer_len, int *offset,
                          const char *arg, const int val)
{/*{{{*/
  int status;
  status = ssnprintf (buffer + *offset, buffer_len - *offset, arg, val);
  if ((status < 1) || (status >= buffer_len - *offset))
     return (-1);
  *offset += status;
  return 0;
}/*}}}*/

static int generate_schema(char *buf, size_t buflen)
{/*{{{*/
  int offset = 0;
  int i;
  for (i = 1; i <= 64; ++i) {
    wrap_ssnprintf_int(buf, buflen, &offset, ", cloud%d DOUBLE", i);
  }
/*
  wrap_ssnprintf_str(buf, buflen, &offset, ", ubuntu DOUBLE", NULL);
*/
  return 0;
}/*}}}*/

static int create_table_host(char *tablename, mysql_database_t *db)
{/*{{{*/
  int offset = 0;
  MYSQL_RES *res;
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "CREATE TABLE IF NOT EXISTS %s ", tablename);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "(time TIMESTAMP DEFAULT 0 PRIMARY KEY", NULL);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "%s", schema_host);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     ")", NULL);
  res = exec_query (db->con, query_str);
  return 0;
}/*}}}*/

static int create_table_mrjob(char *tablename, mysql_database_t *db)
{/*{{{*/
  int offset = 0;
  MYSQL_RES *res;
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "CREATE TABLE IF NOT EXISTS %s ", tablename);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "(time TIMESTAMP, attempt VARCHAR(10), metric VARCHAR(10), value DOUBLE)", NULL);
  res = exec_query (db->con, query_str);
  return 0;
}/*}}}*/

static int mysql_insert_host(char *tablename, time_t time, const char *host,
  const char *value, mysql_database_t *db)
{
  int offset = 0;
  MYSQL_RES *res;
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "INSERT INTO %s ", tablename);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "(time, %s) ", host);
  char buf[80];
  struct tm *ts;
  ts = localtime(&time);
  strftime(buf, sizeof(buf), "\"%Y-%m-%d %H:%M:%S\"", ts);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "VALUE (%s, ", buf);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "%s) ", value);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "ON DUPLICATE KEY UPDATE %s", host);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "=%s", value);
  res = exec_query (db->con, query_str);
  return 0;
}

static int mysql_insert_mrjob(char *tablename, time_t time, const char *taskid,
  const char *metric, const char *value, mysql_database_t *db)
{
  int offset = 0;
  MYSQL_RES *res;
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "INSERT INTO %s ", tablename);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "(time, attempt, metric, value) ", NULL);
  char buf[80];
  struct tm *ts;
  ts = localtime(&time);
  strftime(buf, sizeof(buf), "\"%Y-%m-%d %H:%M:%S\"", ts);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "VALUE (%s,", buf);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "\"%s\",", taskid);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "\"%s\",", metric);
  wrap_ssnprintf_str(query_str, MAX_QUERY_LEN, &offset,
     "%s)", value);
  res = exec_query (db->con, query_str);
  return 0;
}

static int mysql_collectd_init (void)
{/*{{{*/
  generate_schema(schema_host, sizeof(schema_host));
  getconnection(&mydb);
  return (0);
} /* int rc_init *//*}}}*/

static int mysql_collectd_write (const data_set_t *ds, const value_list_t *vl,
    user_data_t __attribute__((unused)) *user_data)
{
  char tablename[512];
  char values[512];
  char taskid[TASK_ID_LEN+1];
  int status, flag;

  if (mydb.con == NULL)
  {
    ERROR ("mysql plugin: MYDB->CON == NULL.");
    plugin_unregister_write ("mysql");
    return (-1);
  }

  if (strcmp(vl->plugin, "mrjob_task") == 0) {
    flag = 0;
  } else
  if (strcmp(vl->plugin, "hadoop") == 0) {
    flag = 1;
  } else {
    return -1;
  }

  if (strcmp (ds->type, vl->type) != 0)
  {
    ERROR ("mysql plugin: DS type does not match value list type");
    return (-1);
  }

  if (value_list_to_tablename (tablename, sizeof (tablename), ds, vl) != 0)
  {
    ERROR ("mysql plugin: value_list_to_filename failed.");
    return (-1);
  }

  if (value_list_to_string (values, sizeof (values), ds, vl) != 0)
  {
    ERROR ("mysql plugin: value_list_to_string failed.");
    return (-1);
  }

  if (flag == 0) {
    status = create_table_mrjob(tablename, &mydb);
    if (status != 0)
    {
      ERROR ("mysql plugin: create (%s) failed", tablename);
      return (-1);
    }
    strcpyn(taskid, vl->plugin_instance + JOB_ID_LEN + 1, TASK_ID_LEN);
    INFO("MRJOB METRIC IS %s", vl->type_instance);
    status = mysql_insert_mrjob(tablename, vl->time, taskid,
      vl->type_instance, values, &mydb);
    if (status != 0)
    {
      ERROR ("mysql plugin: mysql_update (%s, [%s]), %d failed",
          tablename, values, status);
      return (-1);
    }
  } else {
    status = create_table_host(tablename, &mydb);
    if (status != 0)
    {
      ERROR ("mysql plugin: create (%s) failed", tablename);
      return (-1);
    }
    status = mysql_insert_host(tablename, vl->time, vl->host, values, &mydb);
    if (status != 0)
    {
      ERROR ("mysql plugin: mysql_update (%s, [%s]), %d failed",
          tablename, values, status);
      return (-1);
    }
  }

  return (0);
} /* int rc_write */

static int mysql_collectd_shutdown (void)
{
  mysql_database_free(&mydb);
  return (0);
} /* int rc_shutdown */

void module_register (void)
{
  /* initialize all the pointers */
  mydb.host     = NULL;
  mydb.user     = NULL;
  mydb.pass     = NULL;
  mydb.database = NULL;
  mydb.socket   = NULL;
  mydb.con      = NULL;

  plugin_register_config ("mysql", mysql_collectd_config,
    mysql_config_keys, mysql_config_keys_num);
  plugin_register_init ("mysql", mysql_collectd_init);
  plugin_register_write ("mysql", mysql_collectd_write, /* user_data = */ NULL);
  plugin_register_shutdown ("mysql", mysql_collectd_shutdown);
} /* void module_register */
